﻿/// Copyright © 2013 Microsoft Corporation. 
/// This project is released under the Microsoft Public License (MS-PL)
///
namespace DataMorph.Processing
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.IO;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    using DataMorph.Common;
    using DataMorph.Common.Contracts;
    using DataMorph.Common.DataAccess;
    using DataMorph.Common.Entities;
    using DataMorph.Common.Storage;

    using Microsoft.Practices.Unity;
    using Microsoft.WindowsAzure.ServiceRuntime;
    using Microsoft.WindowsAzure.Storage.Blob;

    public class WorkProcessor : IWorkProcessor
    {
        private readonly IWorkQueue workQueue;

        private readonly ITagDataFileParser tagDataFileParser;

        private readonly IDataAccess dataAccess;

        private readonly IProgressNotifier progressNotifier;

        private readonly IDictionary<string, ITagDataWriter> writers = new Dictionary<string, ITagDataWriter>();

        private readonly string deploymentId;

        private readonly string instanceId;

        private ConcurrentBag<ParseTimings> parseTimings = new ConcurrentBag<ParseTimings>();

        private ConcurrentBag<WriteTimings> writeTimings = new ConcurrentBag<WriteTimings>();

        private int lastBatchId;

        [InjectionConstructor]
        public WorkProcessor(
            IWorkQueue workQueue,
            IDataAccess dataAccess,
            ITagDataFileParser tagDataFileParser,
            IProgressNotifier progressNotifier)
            : this(
                workQueue,
                dataAccess,
                tagDataFileParser,
                progressNotifier,
                RoleEnvironment.DeploymentId,
                RoleEnvironment.CurrentRoleInstance.Id)
        {
        }

        public WorkProcessor(
                        IWorkQueue workQueue,
                        IDataAccess dataAccess,
                        ITagDataFileParser tagDataFileParser,
                        IProgressNotifier progressNotifier,
                        string deploymentId,
                        string instanceId)
        {
            if (workQueue == null)
            {
                throw new ArgumentNullException("workQueue", "Work queue must not be null.");
            }

            if (dataAccess == null)
            {
                throw new ArgumentNullException("dataAccess", "Data access must not be null.");
            }

            if (tagDataFileParser == null)
            {
                throw new ArgumentNullException("tagDataFileParser", "Tag data file parser must not be null.");
            }

            if (progressNotifier == null)
            {
                throw new ArgumentNullException("progressNotifier", "Progress notifier must not be null");
            }

            this.workQueue = workQueue;
            this.dataAccess = dataAccess;
            this.tagDataFileParser = tagDataFileParser;
            this.progressNotifier = progressNotifier;
            this.deploymentId = deploymentId;
            this.instanceId = instanceId;
        }

        public void Run()
        {
            while (true)
            {
                WorkRequestBatch batch = null;
                try
                {
                    batch = this.GetNextBatch();
                }
                catch (Exception exception)
                {
                    Trace.TraceError("Exception while getting processing batch {0}", exception);
                }

                if (batch == null && this.parseTimings.Count > 0)
                {
                    this.lastBatchId = 0;
                    this.WriteProcessingRecord();
                    this.parseTimings = new ConcurrentBag<ParseTimings>();
                    this.writeTimings = new ConcurrentBag<WriteTimings>();
                }
                else if (batch == null)
                {
                    Thread.Sleep(TimeSpan.FromSeconds(10));
                }
                else
                {
                    try
                    {
                        this.ProcessRequestBatch(batch, this.GetTagDataWriter);
                        this.writers.Clear();
                    }
                    catch (AggregateException e)
                    {
                        Trace.TraceError("An exception occured while processing work request batch: {0}", e.InnerException.ToString());
                    }
                    catch (Exception e)
                    {
                        Trace.TraceError("An exception occured while processing work request batch: {0}", e.ToString());
                    }
                }
            }
        }

        private void ProcessRequestBatch(WorkRequestBatch batch, Func<ICloudStorageAccount, string, int, ProcessRequest, ITagDataWriter> writerFactory)
        {
            var storageAccounts = new ConcurrentDictionary<string, ICloudStorageAccount>();
            foreach (var storageAccountDetails in this.dataAccess.GetStorageAccountNames())
            {
                storageAccounts.TryAdd(
                                storageAccountDetails.Name,
                                this.dataAccess.GetStorageAccount(storageAccountDetails.Name, storageAccountDetails.Key));
            }

            Parallel.ForEach(
                        batch.WorkRequests,
                        workRequest =>
                        {
                            ICloudStorageAccount storageAccount;
                            if (!storageAccounts.TryGetValue(workRequest.StorageAccountName, out storageAccount))
                            {
                                return;
                            }

                            var processResult = this.ProcessWorkRequest(workRequest, storageAccount, writerFactory, batch.ProcessRequest, batch.Id);
                            if (processResult != null)
                            {
                                this.parseTimings.Add(processResult);
                            }
                        });

            Parallel.ForEach(this.writers.Values, writer => this.writeTimings.Add(writer.Commit()));

            foreach (var workRequest in batch.WorkRequests)
            {
                this.progressNotifier.NotifyFileProcessed(
                    workRequest.File, workRequest.StorageAccountName, workRequest.DataSetName, 0);
            }
        }

        private ParseTimings ProcessWorkRequest(
                                ProcessingWorkRequest workRequest,
                                ICloudStorageAccount storageAccount,
                                Func<ICloudStorageAccount, string, int, ProcessRequest, ITagDataWriter> writerFactory,
                                ProcessRequest processRequest,
                                int batchId)
        {
            try
            {
                var blobClient = storageAccount.CreateCloudBlobClient();
                var container = blobClient.GetContainerReference(workRequest.DataSetName);

                var result = this.ParseFile(workRequest, container);

                foreach (var tagId in result.TagData.Keys)
                {
                    var writer = writerFactory(storageAccount, tagId, batchId, processRequest);
                    writer.AddTagData(result.TagData[tagId]);
                }

                workRequest.MarkRequestProcessed();

                return new ParseTimings
                {
                    StorageAccount = workRequest.StorageAccountName,
                    DataSet = processRequest.DataSetName,
                    FileLength = result.Length,
                    ReadTime = result.ReadTime,
                    ParseTime = result.ParseTime,
                    TimeRequested = processRequest.TimeRequested.UtcDateTime
                };
            }
            catch (AggregateException e)
            {
                Trace.TraceError("Failed to process work request '{0}':{1}", workRequest, e.InnerException);
            }
            catch (Exception e)
            {
                Trace.TraceError("Failed to process work request '{0}':{1}", workRequest, e);
            }

            return null;
        }

        private FileParseResult ParseFile(ProcessingWorkRequest workRequest, CloudBlobContainer container)
        {
            var blockBlob = container.GetBlockBlobReference(workRequest.File);

            using (var memoryStream = new MemoryStream())
            {
                var stopwatch = new Stopwatch();
                stopwatch.Start();
                var stream = blockBlob.OpenRead(
                                            null,
                                            new BlobRequestOptions
                                                {
                                                    MaximumExecutionTime = TimeSpan.FromHours(1),
                                                    RetryPolicy = DataMorphGlobalRetryPolicy.NewPolicy
                                                });

                stream.CopyTo(memoryStream);
                stopwatch.Stop();
                var readTime = stopwatch.Elapsed.TotalSeconds;

                stopwatch.Reset();
                stopwatch.Restart();
                memoryStream.Seek(0, SeekOrigin.Begin);
                var tagData = this.tagDataFileParser.Parse(memoryStream);
                stopwatch.Stop();

                stream.Close();

                return new FileParseResult
                           {
                               TagData = tagData,
                               Length = stream.Length,
                               ReadTime = readTime,
                               ParseTime = stopwatch.Elapsed.TotalSeconds
                           };
            }
        }

        private WorkRequestBatch GetNextBatch()
        {
            var processingWorkRequests = this.workQueue.GetWorkRequests().ToList();
            if (processingWorkRequests.Count == 0)
            {
                return null;
            }

            var workRequest = processingWorkRequests.First();
            return new WorkRequestBatch
            {
                Id = this.lastBatchId++,
                ProcessRequest = new ProcessRequest
                {
                    DataSetName = workRequest.DataSetName,
                    StorageAccountName = workRequest.StorageAccountName,
                    TimeRequested = workRequest.TimeRequested
                },
                WorkRequests = processingWorkRequests.Where(x => x.StorageAccountName.Equals(workRequest.StorageAccountName))
                                                     .Where(x => x.DataSetName.Equals(workRequest.DataSetName)).ToList()
            };
        }

        private ITagDataWriter GetTagDataWriter(ICloudStorageAccount cloudStorageAccount, string tagId, int batchId, ProcessRequest processRequest)
        {
            lock (this.writers)
            {
                if (!this.writers.ContainsKey(tagId))
                {
                    this.writers[tagId] = new TagDataWriter(cloudStorageAccount, dataAccess, tagId, processRequest, this.deploymentId, this.instanceId, batchId);
                }

                return this.writers[tagId];
            }
        }

        private void WriteProcessingRecord()
        {
            var groupedProcessingResults = this.parseTimings.GroupBy(i => i.TimeRequested)
                                               .Select(
                                                   g =>
                                                   new ProcessingRecordEntity
                                                   {
                                                       TimeRequested = g.Key,
                                                       StorageAccount = g.Max(t => t.StorageAccount),
                                                       DataSet = g.Max(t => t.DataSet),
                                                       DeploymentId = RoleEnvironment.DeploymentId,
                                                       InstanceId = RoleEnvironment.CurrentRoleInstance.Id,
                                                       AverageParseTime = g.Average(t => t.ParseTime),
                                                       MaxParseTime = g.Max(t => t.ParseTime),
                                                       MinParseTime = g.Min(t => t.ParseTime),
                                                       AverageReadTime = g.Average(t => t.ReadTime),
                                                       MaxReadTime = g.Max(t => t.ReadTime),
                                                       MinReadTime = g.Min(t => t.ReadTime),
                                                       FilesRead = g.Count()
                                                   });

            var groupedWriteResults = this.writeTimings.GroupBy(i => i.TimeRequested)
                            .Select(
                                g =>
                                new ProcessingRecordEntity
                                {
                                    TimeRequested = g.Key,
                                    DataSet = g.Max(t => t.DataSet),
                                    MaxWriteTime = g.Max(t => t.WriteTime),
                                    MinWriteTime = g.Min(t => t.WriteTime),
                                    AverageWriteTime = g.Average(t => t.WriteTime),
                                    CompletedTime = g.Max(t => t.CompletedTime),
                                    FilesWritten = g.Count()
                                });

            var combined = from processingRecord in groupedProcessingResults
                           join writeRecord in groupedWriteResults on
                               new { processingRecord.DataSet, processingRecord.TimeRequested } equals
                               new { writeRecord.DataSet, writeRecord.TimeRequested }
                           select
                               new ProcessingRecordEntity
                               {
                                   TimeRequested = processingRecord.TimeRequested,
                                   StorageAccount = processingRecord.StorageAccount,
                                   DataSet = processingRecord.DataSet,
                                   DeploymentId = processingRecord.DeploymentId,
                                   InstanceId = processingRecord.InstanceId,
                                   AverageParseTime = Math.Round(processingRecord.AverageParseTime, 2),
                                   MaxParseTime = Math.Round(processingRecord.MaxParseTime, 2),
                                   MinParseTime = Math.Round(processingRecord.MinParseTime, 2),
                                   MaxWriteTime = Math.Round(writeRecord.MaxWriteTime, 2),
                                   MinWriteTime = Math.Round(writeRecord.MinWriteTime, 2),
                                   AverageWriteTime = Math.Round(writeRecord.AverageWriteTime, 2),
                                   CompletedTime = writeRecord.CompletedTime,
                                   TotalProcessingTime = Math.Round((writeRecord.CompletedTime - processingRecord.TimeRequested).TotalSeconds, 0),
                                   AverageReadTime = Math.Round(processingRecord.AverageReadTime, 2),
                                   MaxReadTime = Math.Round(processingRecord.MaxReadTime),
                                   MinReadTime = Math.Round(processingRecord.MinReadTime),
                                   FilesRead = processingRecord.FilesRead,
                                   FilesWritten = writeRecord.FilesWritten,
                                   InstanceCount = RoleEnvironment.CurrentRoleInstance.Role.Instances.Count,
                                   CoreCount = Environment.ProcessorCount,
                                   BatchSize = this.workQueue.WorkBatchSize
                               };

            this.dataAccess.AddProcessingRecords(combined);
        }
    }
}